package netty.common.network;

import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;

import java.util.concurrent.atomic.AtomicInteger;

public class Message {
	private final static AtomicInteger RequestId = new AtomicInteger(1);
	// frame ==> control:[ magic + version + type + reqType + seqId + headLength] [head] [body]
	public final static int MAGIC = 0xf11f;
	public final static int CONTROL_LENGTH = 2 + 1 + 1 + 2 + 4 + 4;

	private short magic = (short) MAGIC;
	private byte version = (byte)10;
	private byte type = 0;    // CALL | REPLY | EXCEPTION
	private short reqHandlerType = 0; // FETCH | PRODUCER
	private int seqId = 0;
	private int headLength = 0;

	private transient byte[] head;
	private transient ByteBuf body;

	public static Message newRequestMessage() {
		Message msg = new Message();
		msg.setType(TransferType.CALL.value);
		msg.setSeqId(RequestId.getAndIncrement());
		return msg;
	}

	public static Message newResponseMessage() {
		Message msg = new Message();
		msg.setType(TransferType.REPLY.value);
		return msg;
	}

	public static Message newExceptionMessage() {
		Message msg = new Message();
		msg.setType(TransferType.EXCEPTION.value);
		return msg;
	}

	public static Message newOneWayMessage() {
		Message msg = new Message();
		msg.setType(TransferType.ONEWAY.value);
		msg.setSeqId(RequestId.getAndIncrement());
		return msg;
	}

	public short getMagic() {
		return magic;
	}

	public byte getVersion() {
		return version;
	}

	public void setVersion(byte version) {
		this.version = version;
	}

	public byte getType() {
		return type;
	}

	public void setType(byte type) {
		this.type = type;
	}

	public Message setSeqId(int seqId) {
		this.seqId = seqId;
		return this;
	}

	public int getSeqId() {
		return seqId;
	}
	
	public short getReqHandlerType() {
		return reqHandlerType;
	}

	public void setReqHandlerType(short reqHandlerType) {
		this.reqHandlerType = reqHandlerType;
	}

	public int getHeadLength() {
		return headLength;
	}

	public void setHeadLength(int headLength) {
		this.headLength = headLength;
	}

	public byte[] getHead() {
		return head;
	}

	public void setHead(byte[] head) {
		this.head = head;
	}

	public void setBody(ByteBuf body) {
		this.body = body;
	}

	public ByteBuf getBody() {
		return body;
	}

	public int serializedSize() {
        return Message.CONTROL_LENGTH + headLength() + bodyLength();
    }

	public int headLength() {
		return this.head == null ? 0 : this.head.length;
	}

	public int bodyLength() {
		return this.body == null ? 0 : this.body.readableBytes();
	}
	
	public void writeToByteBuf(ByteBuf byteBuf){
		if(byteBuf != null){
			// totalLength =  CONTROL_LENGTH + head + body.
			byteBuf.writeInt(serializedSize());

			// control
			byteBuf.writeShort(getMagic());
			byteBuf.writeByte(getVersion());
			byteBuf.writeByte(getType());
			byteBuf.writeShort(getReqHandlerType());
			byteBuf.writeInt(getSeqId());
			byteBuf.writeInt(getHead().length);

			// head
			if (headLength() > 0) {
				byteBuf.writeBytes(Unpooled.wrappedBuffer(getHead()));
			}

			// body
			if (bodyLength() > 0) {
				byteBuf.writeBytes(getBody());
			}

		}
	}
	
	public void readFromByteBuf(ByteBuf byteBuf){
		if (byteBuf != null) {
			// 读取头信息.
			int magic = byteBuf.readUnsignedShort();
			if (magic != Message.MAGIC) {
				byteBuf.discardReadBytes();
				byteBuf.release();
				throw new RuntimeException("协议Magic无效!");
			}
			byte version = byteBuf.readByte();
			byte type = byteBuf.readByte();
			short reqHandlerType = byteBuf.readShort();
			int seqId = byteBuf.readInt();
			int headLength = byteBuf.readInt();

			// 读取head
			byte[] head = new byte[headLength];
			byteBuf.readBytes(head);

			// 读取body
			body = byteBuf.retainedSlice().duplicate();

			this.setVersion(version);
			this.setType(type);
			this.setSeqId(seqId);
			this.setReqHandlerType(reqHandlerType);
			this.setHeadLength(headLength);
			this.setHead(head);
			this.setBody(body);
		}
	}
	
	public static Message buildFromByteBuf(ByteBuf byteBuf){
		Message message = new Message();
		message.readFromByteBuf(byteBuf);
		return message;
	}

	@Override
	public String toString() {
		return String.format("magic:%s, version:%s, type:%s, reqHandlerType:%s, seqId:%d", magic, version, type, reqHandlerType, seqId);
	}
	
	@Override
    public boolean equals(Object obj) {
        if (obj instanceof Message) {
            Message m = (Message) obj;
            return getMagic() == m.getMagic()//
                    && getVersion() == m.getVersion()//
                    && getType() == m.getType()//
                    && getReqHandlerType() == m.getReqHandlerType()//
                    && getSeqId() == m.getSeqId()
                    && getHeadLength() == m.getHeadLength()
                    && headLength() == m.headLength()
                    && bodyLength() == m.bodyLength();
        }
        return false;
    }

    
}
